package com.example.mq.mqserver.datacenter.Impl;

import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.Binding;
import com.example.mq.mqserver.core.Exchange;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;
import com.example.mq.mqserver.datacenter.IDiskDataCenter;
import com.example.mq.mqserver.datacenter.IMemoryDataCenter;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class MemoryDataCenterImpl implements IMemoryDataCenter {
    //  在内存中操作数据，先准备数据结构
    //  在此使用 Map 为数据结构，注意线程安全
    //  key=exchangeName, value=exchange
    private final ConcurrentHashMap<String,Exchange> exchangeMap = new ConcurrentHashMap<>();
    //  key=queueName, value=queue
    private final ConcurrentHashMap<String,MSGQueue> queueMap = new ConcurrentHashMap<>();
    //  key=exchangeName, value=(key=queueName,value=Binding)
    private final ConcurrentHashMap<String,ConcurrentHashMap<String,Binding>> bindingsMap = new ConcurrentHashMap<>();
    //  所有消息
    //  key=messageId, value=message
    private final ConcurrentHashMap<String,Message> messageMap = new ConcurrentHashMap<>();
    //  队列里的消息（消息队列本体）
    //  key=queueName, value=messageList
    private final ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    //  待确认消息
    //  key=queueName,value=(key=messageId,value=message)
    private final ConcurrentHashMap<String,ConcurrentHashMap<String,Message>> messageWaitAckMap = new ConcurrentHashMap<>();



    @Override
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(),exchange);
    }
    @Override
    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }
    @Override
    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
    }


    @Override
    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(),queue);
    }
    @Override
    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }
    @Override
    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
    }


    @Override
    public void insertBinding(Binding binding) throws MqException {
        String exchangeName = binding.getExchangeName();
        String queueName = binding.getQueueName();
        //  判断当前的交换机是否已经有哈希表，如果没有就创建
        ConcurrentHashMap<String,Binding> bindingMap =
                bindingsMap.computeIfAbsent(exchangeName, v -> new ConcurrentHashMap<>());
        //  再判断当前交换机和队列的绑定是否已经存在
        if(bindingMap.get(queueName) != null) {
            throw new MqException("当前绑定已存在! ");
        }
        bindingMap.put(queueName,binding);
    }
    @Override
    public Binding getBinding(String exchangeName, String queueName) {
        Map<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if(bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
    }
    @Override
    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }
    @Override
    public void deleteBinding(Binding binding) {
        String exchangeName = binding.getExchangeName();
        String queueName = binding.getQueueName();
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if(bindingMap == null) {
            //  该交换机没有任何绑定
            return;
        }
        bindingMap.remove(queueName);
    }


    @Override
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(),message);
    }
    @Override
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }
    @Override
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
    }


    @Override
    public void offerMessage(String queueName, Message message) {
        //  判断队列对应的链表是否存在，不存在就创建
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queueName,v -> new LinkedList<>());
        synchronized (messages) {
            messages.add(message);
        }
        messageMap.put(message.getMessageId(),message);
    }
    @Override
    public Message pollMessage(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages == null) {
            return null;
        }
        synchronized (messages) {
            //  链表头删
            return messages.poll();
        }
    }
    @Override
    public int getQueueSize(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages == null) {
            return 0;
        }
        synchronized (messages) {
            return messages.size();
        }
    }


    @Override
    public void addMessageWaitAck(String queueName, Message message) {
        ConcurrentHashMap<String, Message> messageMap =
                messageWaitAckMap.computeIfAbsent(queueName, v -> new ConcurrentHashMap<>());
        messageMap.put(message.getMessageId(),message);
    }
    @Override
    public Message removeMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageMap = messageWaitAckMap.get(queueName);
        if(messageMap == null) {
            return null;
        }
        return messageMap.remove(messageId);
    }


    @Override
    public void recovery(IDiskDataCenter diskDataCenter) throws IOException, ClassNotFoundException {
        //  恢复交换机
        List<Exchange> exchanges = diskDataCenter.selectAllExchange();
        for(Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(),exchange);
        }
        //  恢复队列
        List<MSGQueue> queues = diskDataCenter.selectAllQueue();
        for(MSGQueue queue : queues) {
            queueMap.put(queue.getName(),queue);
        }
        //  恢复绑定
        List<Binding> bindings = diskDataCenter.selectAllBinding();
        for(Binding binding : bindings) {
            String exchangeName = binding.getExchangeName();
            String queueName = binding.getQueueName();
            ConcurrentHashMap<String, Binding> bindingMap =
                    bindingsMap.computeIfAbsent(exchangeName, v -> new ConcurrentHashMap<>());
            bindingMap.put(queueName,binding);
        }
        //  恢复消息
        for(MSGQueue queue : queues) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            //  恢复队列的消息
            queueMessageMap.put(queue.getName(),messages);

            //  恢复所有消息
            for(Message message : messages) {
                messageMap.put(message.getMessageId(),message);
            }
        }

    }

}
